package com.bing.test;

import org.apache.activemq.broker.BrokerService;
import org.fusesource.mqtt.client.*;
import org.junit.Before;
import org.junit.BeforeClass;
import org.junit.Test;

import java.util.concurrent.TimeUnit;

/**
 * Created by Administrator on 2015/7/1.
 */
public class HelloMqtt {
    private static final String TOPIC = "helloMQTT";
    private static final String HELLO_MQTT = "hello MQTT";
    private static final String LOCALHOST = "localhost";
    private static final int PORT = 1883;

    FutureConnection connection;
    FutureConnection connections;

    @BeforeClass
    public static void beforeCass() throws Exception {
        BrokerService broker = new BrokerService();
        broker.setBrokerName("testName");//如果启动多个Broker时，必须为Broker设置一个名称
        broker.addConnector("mqtt://localhost:" + PORT);
        broker.start();
    }

    @Before
    public void before() throws Exception {

        MQTT mt = new MQTT();
        mt.setHost(LOCALHOST, PORT);
        connection = mt.futureConnection();
        connection.connect();
        connection.subscribe(new Topic[]{new Topic(TOPIC, QoS.AT_LEAST_ONCE)});
        //mt.setKeepAlive(KEEP_ALIVE);
        System.out.println("before");
        MQTT mqtt = new MQTT();
        mqtt.setHost(LOCALHOST, PORT);

        connections = mqtt.futureConnection();
        connections.connect();

    }

    @Test
    public void HelloMqtt() throws Exception {

        Future<Message> futrueMessage = connection.receive();
        Message message = null;
        try {
              message = futrueMessage.await(1, TimeUnit.SECONDS);
        } catch (Exception e) {
            System.out.println(e);
        }


         connections.publish(TOPIC, HELLO_MQTT.getBytes(), QoS.AT_LEAST_ONCE,
                true);
        connections.publish(TOPIC, "oh yeash".getBytes(), QoS.AT_LEAST_ONCE,
                true);
        futrueMessage = connection.receive();
        message = futrueMessage.await(1, TimeUnit.SECONDS);


        System.out.println("MQTTFutureClient.Receive Message " + "Topic Title :" + message.getTopic() + " context :" + String.valueOf(message.getPayloadBuffer()));

        futrueMessage = connection.receive();
        message = futrueMessage.await(1, TimeUnit.SECONDS);


        System.out.println("MQTTFutureClient.Receive Message " + "Topic Title :" + message.getTopic() + " context :" + String.valueOf(message.getPayloadBuffer()));
    }
}
